package com.zhaokuner.springcloudconfigserver.zk;

import lombok.Data;
import lombok.extern.log4j.Log4j2;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.text.MessageFormat;
import java.util.LinkedHashMap;
import java.util.List;

/**
 * <p>
 * Description:zkClient
 * <p>
 * Copyright: 2018-现在 北京掌通未来科技有限公司
 * <p>
 *
 * @author 赵希坤 zhaoxikun
 * @date 2020/7/25 14:36
 */
@Component
@Log4j2
@Data
public class CuratorClient implements ApplicationRunner {
    /**
     * zk说明<p>
     * zookeeper.extendedTypesEnabled ：(仅 Java 系统 property：zookeeper.extendedTypesEnabled)
     * **3.5.4 的新功能：**定义为“ true”以启用扩展的 features(例如创建 TTL 节点)。
     * 默认情况下禁用它们。重要信息：由于内部限制，启用的服务器 ID 必须小于 255。
     * <p>
     * zookeeper.emulate353TTLNodes ：(仅 Java 系统 property：zookeeper.emulate353TTLNodes)
     * 3.5.4 的新功能：由于 ZOOKEEPER-2901，3.5.4/3.6.0 不支持在 version 3.5.3 中创建的 TTL 节点。
     * 但是，通过 zookeeper.emulate353TTLNodes 系统 property 提供了解决方法。
     * 如果您在 ZooKeeper 3.5.3 中使用了 TTL 节点，并且除了zookeeper.extendedTypesEnabled之外，还需要将兼容性zookeeper.emulate353TTLNodes设置为“ true”。
     * 注意：由于该错误，服务器 ID 必须小于或等于 127。此外，最大支持 TTL value 是 1099511627775，小于 3.5.3 中允许的值(1152921504606846975)
     */
    private CuratorFramework client = null;

    @Value("${spring.application.name}")
    private String configServerName = "spring_cloud_config_server";
    @Value("${spring.cloud.config.zkConnectString}")
    private String zkConnectString;
    @Value("${spring.cloud.config.zkChangeLog.enabled}")
    private Boolean zkChangeLogEnabled;
    @Value("${spring.cloud.config.zkChangeLog.prefix}")
    private String zkChangeLogPrefix;
    @Value("${spring.cloud.config.zkChangeLog.minCount}")
    private int zkChangeLogMinCount;
    @Value("${spring.cloud.config.zkChangeLog.maxCount}")
    private int zkChangeLogMaxCount;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        init();
        log.info("********************CuratorClient config over");
    }

    public void init() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                .connectString(zkConnectString)
                .retryPolicy(retryPolicy)
                .build();
        client.start();
        log.info("********************CuratorClient start success");
    }

    public Object change(String branch, String fileName, String value) throws Exception {
        String path = MessageFormat.format("/{0}/{1}/{2}", configServerName, branch, fileName);
        // 这里就算分成两步，保存再更新，client端还是仅收一个create事件
        // 除非两个操作间sleep 下，先让create事件触发后再update
        client.create().orSetData().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, value.getBytes(StandardCharsets.UTF_8));
        log.info("********************change_branch_file label:{} fileName:{} value:{} ", branch, fileName, value);
        changeLog(branch, fileName, value);
        return path;
    }

    public void changeLog(String branch, String fileName, String value) {
        //是否开启添加
        if (!Boolean.TRUE.equals(zkChangeLogEnabled)) {
            return;
        }
        try {
            String path = MessageFormat.format("/{0}/{1}/{2}-", zkChangeLogPrefix, branch, fileName);
            client.create().orSetData().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, value.getBytes(StandardCharsets.UTF_8));
            Stat stat = new Stat();
            List<String> list = client.getChildren().storingStatIn(stat).forPath(MessageFormat.format("/{0}/{1}", zkChangeLogPrefix, branch));
            if (stat.getNumChildren() >= zkChangeLogMaxCount) {
                log.info("********************change_log_delete count:{} max:{} min:{} ", list.size(), zkChangeLogMaxCount, zkChangeLogMinCount);
                list.stream().sorted((o, n) -> {
                    //倒排 从大到小
                    return Long.parseLong(o.substring(o.lastIndexOf("-") + 1)) - Long.parseLong(n.substring(n.lastIndexOf("-") + 1)) < 0 ? 1 : -1;
                }).skip(zkChangeLogMinCount).forEach(a -> {
                    try {
                        client.delete().inBackground().forPath(MessageFormat.format("/{0}/{1}/{2}", zkChangeLogPrefix, branch, a));
                    } catch (Exception e) {
                        log.error(e);
                    }
                });
            }
        } catch (Throwable e) {
            log.error(e);
        }
    }

    public List<String> getRootChildren() throws Exception {
        List<String> list = client.getChildren().forPath(MessageFormat.format("/{0}", zkChangeLogPrefix));
        return list;
    }

    public LinkedHashMap<String, String> getBranchChildren(String label) throws Exception {
        List<String> list = client.getChildren().forPath(MessageFormat.format("/{0}/{1}", zkChangeLogPrefix, label));
        list.sort((o, n) -> {
            //倒排 从大到小
            return Long.parseLong(o.substring(o.lastIndexOf("-") + 1)) - Long.parseLong(n.substring(n.lastIndexOf("-") + 1)) < 0 ? 1 : -1;
        });

        LinkedHashMap<String, String> map = new LinkedHashMap<>(8);
        for (String s : list) {
            String str = new String(client.getData().forPath(MessageFormat.format("/{0}/{1}/{2}", zkChangeLogPrefix, label, s)));
            map.put(s, str);
        }
        return map;
    }

    @PreDestroy
    public void close() {
        closeClient();
    }

    private void closeClient() {
        if (client != null) {
            client.close();
            log.info("********************CuratorClient close");
        }
    }

}
